package com.adcc.monitoragent.core.service;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.PrintStream;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import org.apache.commons.net.telnet.TelnetClient;
import org.quartz.DisallowConcurrentExecution;
import org.quartz.Job;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;
import org.quartz.PersistJobDataAfterExecution;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.adcc.monitoragent.bean.ZookeeperBean;
import com.adcc.monitoragent.log.ServiceLogger;
import com.adcc.monitoragent.send.MessageSend;
import com.adcc.monitoragent.util.Constants;
import com.adcc.monitoragent.util.ProtertiesUtil;

/**
 * ip + port 唯一确定 集群监控
 * 
 * @author charlie
 * 
 */
@PersistJobDataAfterExecution   
@DisallowConcurrentExecution
public class ZookeeperService implements Job {
	private final static Logger log = LoggerFactory.getLogger(ZookeeperService.class);
	private static Map<String, ZookeeperBean> lastZookeeperBeanMap = new ConcurrentHashMap<String, ZookeeperBean>();

	public static ZookeeperBean getInfo(String ip, int port,
			String username, String password) {
		ZookeeperBean zkBean = new ZookeeperBean();
		zkBean.setIp(ip);
		zkBean.setPort(port + "");
		//默认异常状态
		zkBean.setStatus(Constants.STATUS_ERROR);
		try {
			String status = cmd(ip, port, "ruok");
			String temp = null;
			if (!"imok".equals(status)) {
				//重连一次
				Thread.sleep(1000);
				status = cmd(ip, port, "ruok");
			}
			if ("imok".equals(status)) {
				zkBean.setStatus(Constants.STATUS_OK);
				temp = cmd(ip, port, "stat");
				if(!"".equals(temp)) {
					String[] stat = temp.split("\n");
					// 版本
					zkBean.setVersion(stat[0].split(": ")[1].split(",")[0]);
					int index = -1;
					for (int i = 4; i < stat.length; i++) {
						if (index == -1) {
							if (stat[i].startsWith("Latency")) {
								index = i;
							}
						}
					}
					// 获取client
					for (int i = 2; i < index - 1; i++)
						zkBean.getClients().add(stat[i].substring(1));
					// 其他数据
					zkBean.setLatency(stat[index].split(": ")[1]);
					zkBean.setReceived(stat[index + 1].split(": ")[1]);
					zkBean.setSent(stat[index + 2].split(": ")[1]);
					zkBean.setOutstanding(stat[index + 3].split(": ")[1]);
					String state = "";
					if (stat[index + 5].split(": ")[1].equals("follower")) {
						state = Constants.STANDBY;
					} else {
						state = Constants.MASTER;
					}
					zkBean.setState(state);
				zkBean.setCount(stat[index + 6].split(": ")[1]);
				}
			} 
		} catch (Exception e) {
			e.printStackTrace();
		}
		return zkBean;
	}

	private static String cmd(String ip, int port, String s) {
		StringBuilder str = new StringBuilder();
		BufferedReader in= null;
		PrintStream out = null;
		TelnetClient telnet = null;
		try {
			telnet = new TelnetClient();
			telnet.setDefaultTimeout(10000);
			telnet.setConnectTimeout(10000);
			// 连接
			telnet.connect(ip, port);
			
			//Set the timeout in milliseconds of a currently open connection. Only call this method after a connection has been opened by connect().
			//To set the initial timeout, use setDefaultTimeout(int) instead.
			telnet.setSoTimeout(10000);
			
			in = new BufferedReader(new InputStreamReader(
					telnet.getInputStream()));
			out = new PrintStream(telnet.getOutputStream());
			out.println(s);
			out.flush();
			String buf = "";
			while ((buf = in.readLine()) != null && !buf.equals("null")) {
				str.append(buf).append("\n");
			}
		} catch (Exception e) {
			log.error("zookeeper连接失败:" + e.getMessage());
			e.printStackTrace();
		} finally {
			if (telnet != null) {
				try {
					telnet.disconnect();
				} catch (IOException e1) {
					e1.printStackTrace();
				}
			}
		}
		return str.toString().trim();
	}
	
	@Override
	public void execute(JobExecutionContext arg0) throws JobExecutionException {
		log.debug("定时执行任务。。。。。。。。。。。。");
		for (Object objStr : ProtertiesUtil.getZookeeperConnectString()) {
			try {
				String[] s = objStr.toString().split(":");
				ZookeeperBean zk = ZookeeperService.getInfo(s[0], Integer.parseInt(s[1]),
						ProtertiesUtil.getZookeeperUsername(), ProtertiesUtil.getZookeeperPassword());
				//对比状态，如果状态改变发送事件
				String key = zk.buildKey();
				ZookeeperBean lastZkBean = lastZookeeperBeanMap.get(key);
				boolean statusChange = lastZkBean != null && !lastZkBean.getStatus().equals(zk.getStatus());
				if(zk.getStatus().equals(Constants.STATUS_ERROR) && statusChange) {//如果状态异常 并且 状态改变
					log.warn("Zookeeper状态正常转异常，休眠2秒重试，防止误判");
					try {
						Thread.sleep(2000);
					} catch (InterruptedException e) {
					}
					zk = ZookeeperService.getInfo(s[0], Integer.parseInt(s[1]),
							ProtertiesUtil.getZookeeperUsername(), ProtertiesUtil.getZookeeperPassword());
					if(zk.getStatus().equals(Constants.STATUS_ERROR)) {//状态还是异常
						//String desc = "Zookeeper状态改变:" + lastZkBean.getStatus() + "->" + zk.getStatus();
						String desc = "服务停止";
						ServiceLogger.error(zk.getIp(), zk.getPort(), "zookeeper", desc);
					}
				} else if(statusChange){//状态由异常转正常
					//String desc = "Zookeeper状态改变:" + lastZkBean.getStatus() + "->" + zk.getStatus();
					String desc = "服务启动";
					ServiceLogger.info(zk.getIp(), zk.getPort(), "zookeeper", desc);
				}
				lastZookeeperBeanMap.put(key, zk);
				//转发消息
				MessageSend.sendMsg(zk);
			} catch (Exception e) {
				log.error("Zookeeper定时任务执行失败:" + objStr +  e.getMessage());
			}
		}
		log.debug("完成任务。。。。。。。。。。。。");
	}
	

	public static void main(String[] args) throws Exception {
		
		while(true) {
			ZookeeperBean zk = ZookeeperService.getInfo("192.168.243.77", 2171,
					null, null);
			System.out.println(zk.getIp() + zk.getStatus());
			zk = ZookeeperService.getInfo("192.168.243.78", 2172,
					null, null);
			System.out.println(zk.getIp() + zk.getStatus());
			zk = ZookeeperService.getInfo("192.168.243.249", 2173,
					null, null);
			System.out.println(zk.getIp() + zk.getStatus());
			Thread.sleep(5000);
		}
		
	}

}
